package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.SectorPutHbaseWindowFunction;
import cn.itcast.function.SectorWindowFunction;
import cn.itcast.function.StockSecWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.keyBy.KeySelectFunction;
import cn.itcast.sink.SinkHbase;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * Created by JiaYang Xu
 * Date 2021/7/6 16:29
 * Description
 */
public class SectorSecTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        waterData.keyBy(new KeySelectFunction())
                .timeWindow(Time.seconds(5))
                .apply(new StockSecWindowFunction())
                .timeWindowAll(Time.seconds(5))
                .apply(new SectorWindowFunction())
                .timeWindowAll(Time.seconds(5))
                .apply(new SectorPutHbaseWindowFunction())
                .addSink(new SinkHbase(QuotConfig.config.getProperty("sector.hbase.table.name")));
    }
}
